package com.mashibing.disruptor.v1;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.EventHandlerGroup;
import com.lmax.disruptor.dsl.ProducerType;
import java.nio.ByteBuffer;
import java.time.LocalTime;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class Main2 {
    // 单生产者多消费者
    public static void main(String[] args) throws InterruptedException {

        Disruptor<LongEvent> disruptor = new Disruptor<>(new LongEventFactory(), 16,
                Executors.defaultThreadFactory(), ProducerType.SINGLE, new YieldingWaitStrategy());

        LongEventHandler1 longEventHandler1 = new LongEventHandler1();
        LongEventHandler2 longEventHandler2 = new LongEventHandler2();
        EventHandler<LongEvent> eventHandler3 = (event, sequence, endOfBatch) -> {
            System.out.printf("\033[32m %s ===>> %s ===>> eventHandler3 ===>> %s ===>> %s \033[0m\n",
                    Thread.currentThread().getName(), LocalTime.now(), sequence, event.getValue());
            System.out.println();
        };
        EventHandler<LongEvent> eventHandler4 = (event, sequence, endOfBatch) ->
                    System.out.printf("\033[31m %s ===>> %s ===>> eventHandler4 ===>> %s \033[0m\n",
                            Thread.currentThread().getName(), LocalTime.now(), event.getValue());
        EventHandler<LongEvent> eventHandler5 = (event, sequence, endOfBatch) ->
                    System.out.printf("\033[30m %s ===>> %s ===>> eventHandler5 ===>> %s \033[0m\n",
                            Thread.currentThread().getName(), LocalTime.now(), event.getValue());

//        disruptor.handleEventsWith(longEventHandler1, longEventHandler2, eventHandler3);
////        disruptor.handleEventsWith(longEventHandler2, longEventHandler1);

//        disruptor.handleEventsWith(longEventHandler1);
//        disruptor.handleEventsWith(longEventHandler2);
//        disruptor.handleEventsWith(eventHandler3);

//        disruptor.after(longEventHandler1, longEventHandler2)
//                .handleEventsWith(eventHandler3);

        disruptor.handleEventsWith(longEventHandler1).then(longEventHandler2).then(eventHandler3);

//        EventHandlerGroup<LongEvent> eventEventHandlerGroup = disruptor.handleEventsWith(longEventHandler1, longEventHandler2);
//        eventEventHandlerGroup.then(eventHandler3);

//        disruptor.handleEventsWith(longEventHandler1, longEventHandler2);
//        disruptor.after(longEventHandler1).handleEventsWith(eventHandler4);
//        disruptor.after(longEventHandler2).handleEventsWith(eventHandler5);
//        disruptor.after(eventHandler4, eventHandler5).handleEventsWith(eventHandler3);

        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();

//        SequenceBarrier barrier = ringBuffer.newBarrier();
//        MyEventProcessor customProcessor = new MyEventProcessor(ringBuffer, barrier);
//        disruptor.handleEventsWith(processor);

        disruptor.start();

        LongEventProducer1 producer1 = new LongEventProducer1(ringBuffer);

        ByteBuffer bb = ByteBuffer.allocate(8);
        for(long l = 0; l<=15; l++) { // allocate(8) ===>> l<=100
            bb.putLong(0, l);
            producer1.onData(bb);
//            EventTranslator<LongEvent>[] eventTranslators = new EventTranslator<LongEvent>[2];
//            if (l % 10 == 0) {
//                boolean b = ringBuffer.tryPublishEvents(, 1, 10);
//            }
//            TimeUnit.MILLISECONDS.sleep(5000);
        }

        disruptor.shutdown();
    }
}
